Wraps any Beam UnboundedSource as a Spark 4 DataSourceV2 micro-batch
streaming source. The rows have exactly two columns, payload of type
BINARY holding the element encoded with a FullWindowedValueCoder, and
eventTimestamp of type TIMESTAMP holding the Beam event timestamp, so no
Catalyst encoder has to be generated for Beam types.

Offsets are opaque epoch counters. The driver never reads from the source
and never inspects its progress, latestOffset simply returns the previous
epoch plus one so Spark keeps planning micro-batches. Termination is
therefore owned by the lifecycle, not by the offsets.

The source is split once on the driver and the sub sources are memoized,
so every micro-batch plans the same stable set of partitions. On the
executor a static cache keyed by checkpoint location, source id and split
id keeps the Beam UnboundedReader alive across micro-batches, mirroring
the legacy MicrobatchSource. Checkpoint marks are cached in executor
memory only and are never committed durably, which is documented POC
scope with weak failure recovery.

UnboundedSourceDataset is the translator facing entry point. It applies
withWatermark exactly once, since Spark 4 forbids a second declaration
further down the plan.

The test confirms that the EventTimeWatermark logical node survives one
and two typed maps in both the logical and the analyzed plan, and that
the running query really tracks the watermark afterwards. It also reads a
finite set of elements end to end and checks the decoded payloads and
timestamps.

Adds a single generic transformWithState operator that can host any keyed
Beam transform, both the stateful ParDo stack and the group also by window
stack that implements a windowed GroupByKey. The mode is selected by
BeamStatefulProcessorConfig, everything else is shared.

Keys, inputs and outputs are raw Beam coder bytes and every Spark encoder
involved is BINARY or STRING, so Catalyst never has to derive a schema for a
Beam type. TwsTransformFactory owns the row layout and documents it.

Beam state is bridged by TwsStateInternals on top of BytesKV, a tiny string
to bytes store backed by exactly one Spark MapState. One map is a
requirement rather than a preference, ReduceFnRunner invents state tags at
runtime so the set of state addresses is not known when init has to declare
its state variables. Timers are bridged by TwsTimerInternals, which keeps
the full TimerData in a second map and registers only bare wake ups with
Spark, reconciling them against listTimers so an expiry can never be
registered twice.

Two boundary conditions needed care. Spark deletes the expiry it is firing
after the callback returns, so a wake up re armed at that same millisecond
is nudged one millisecond forward. Spark also expires a wake up as soon as
the expiry is at or before the batch watermark, while Beam only fires an
event time timer once the watermark is strictly past it, so timers sitting
exactly on the watermark are withheld and re armed instead of being handed
to Beam, which would decline to fire them and silently drop the on time
pane.

Processing time timers are rejected with a clear message, transformWithState
runs in a single TimeMode and this operator uses event time.

The state and timer bridges are unit tested against an in memory store, and
BeamStatefulProcessorTest runs both modes through a real streaming query on
a live SparkSession.

Shadows PipelineTranslatorFactory so streaming pipelines dispatch to a new
PipelineTranslatorStreaming, which extends PipelineTranslatorBatch to reuse
Impulse, Window.Assign, Flatten, Reshuffle, the bounded read and stateless
ParDo unchanged, while registering placeholders for the unbounded read,
GroupByKey and stateful ParDo that WS-D2 will replace. Combine.PerKey is
deliberately left unregistered so it auto-expands into GroupByKey plus ParDo.

Adds StreamingEvaluationContext, which starts one noop-sink streaming query
per leaf dataset with a checkpoint directory derived from the pipeline
options, registers an idle-stop listener for test termination, and makes
stop() idempotent and safe to call concurrently with evaluate() so cancel()
can interrupt a running streaming pipeline from another thread.

…etons

Adds StreamingTestUtils, the ListBackedUnboundedSource plus static collector
and pipeline option factories the streaming translators will be tested
against, and five skeleton test classes for the stateless ParDo, windowed
GroupByKey, stateful ParDo, chained stateful, and lifecycle scenarios. Every
skeleton compiles and runs, reporting as skipped: each is Ignore'd with a
comment naming the exact WS-D2 translator it needs once that lands.

Replaces the WS-D2 placeholders in PipelineTranslatorStreaming with real
translators for the unbounded read, GroupByKey and stateful ParDo.

ReadUnboundedTranslator wraps the Beam UnboundedSource with
UnboundedSourceDataset, which already declares the single withWatermark for
the whole query, and decodes the binary payload column into the
Dataset<WindowedValue<T>> shape every other translator consumes. The event
timestamp column is projected away, the EventTimeWatermark plan node survives
below the projection and transformWithState reads the query wide watermark
rather than a column.

GroupByKeyStreamingTranslator and StatefulParDoStreamingTranslator both encode
their input into the byte[] row layout of TwsTransformFactory, hand it to the
generic transformWithState operator in GROUP_ALSO_BY_WINDOW and STATEFUL_PARDO
mode respectively, and decode the tagged output rows back out. The stateful
ParDo splits the output by tag index and skips additional outputs that nothing
consumes, so an unused tag does not start a second streaming query.

StreamingTranslationHelpers holds the shared guards and row conversions.
Pipelines outside the POC scope are rejected at translation time with a message
naming the feature, specifically merging windows, custom triggers,
accumulating panes, processing time timers, non deterministic key coders,
non KV input to a stateful ParDo, side inputs on a stateful ParDo,
@OnWindowExpiration and @RequiresTimeSortedInput.

Turns the five Ignore'd streaming test skeletons into real, asserting
tests against the WS-D2 translators.

Every skeleton declared its pipeline as a local TestPipeline variable,
which TestPipeline itself rejects at run() because it is not a Rule
field. These tests cannot use PAssert anyway, a streaming pipeline here
never gets a final watermark so panes never finalize, and each test
needs its own options, so TestPipeline buys nothing. They now build a
plain Pipeline.create(options) and assert after waitUntilFinish against
the static collector in StreamingTestUtils. The unused TestPipeline
factory is dropped from StreamingTestUtils and the reasoning is recorded
in its javadoc.

StatelessParDoStreamingTest and StreamingPipelineLifecycleTest gain the
same relaxed Kryo SparkSessionRule as the rest of the suite, so that the
whole package shares one session and a cancelled or idle stopped query
cannot take that session down with it.

Asserted results, all derived from each test's own input list.

  Stateless ParDo, exactly 0, 2, 4 up to 18.
  Fixed 10s window count per key, a=2 and b=1.
  Sliding 10s every 5s, two a=2 panes, one per shared window.
  Late data dropped, a=1 and sentinel=1, never a=2.
  Stateful ParDo, a, b, c once each plus two timer sentinels.
  Chained dedup into windowed sum, a=8 and b=10.

The chained case is the one that matters, a=13 would mean the dedup
state was lost and an empty result would mean the watermark got stuck
between the two operators, so exactly a=8 and b=10 is the observable
proof of cross operator watermark propagation.

The late data test is the only one that depends on how the source round
robins elements across splits, so it pins maxRecordsPerMicroBatch to 1,
spells out the resulting per batch schedule in a comment, and asserts
the split count assumption up front rather than leaving it silent.

The lifecycle tests wait for a query to actually be active before
cancelling, since translation is asynchronous and an early cancel would
find a null evaluation context, and they check the query really stopped
rather than only that the state was relabelled.

Spark 4 broadcasts a StateSchemaMetadata to the executors for every
transformWithState query, using the user Kryo instance. Nothing
registered that class, so any Beam streaming pipeline with state or
timers died on its first micro-batch as soon as
spark.kryo.registrationRequired was on, which is the default for this
module's tests and a perfectly reasonable production setting. The same
applies to MemoryWriterCommitMessage, the commit message of Spark's
memory sink, which is nested inside the already registered
DataWritingSparkTaskResult.

Both are registered by name, because the shared runner base also
compiles against Spark 3 where neither class exists, and both are
registered with a JavaSerializer rather than Kryo's default field
serializer. Both are Scala case classes holding further Scala and Spark
types, immutable.Map, StructType, avro Schema and Row, none of which are
registered either, so going through Java serialization covers the whole
object graph at once instead of forcing this list to track Spark's
internal field layout. Neither object is on a hot path.

Every streaming test now runs with the module default of
spark.kryo.registrationRequired=true, the per test relaxations are gone.
SparkKryoRegistratorStreamingTest names both classes at compile time
against the Spark 4 classpath, so a future rename becomes a compile
error rather than a silently dropped registration.

Two javadoc paragraphs in the shared base drifted from google-java-format,
one from the streaming note added to SparkStructuredStreamingRunner and one
from the new Kryo registrations. The shared base sources under
runners/spark/src are checked by the :runners:spark project rather than by
:runners:spark:3 or :runners:spark:4, whose own spotless targets only see
their per version override directories, so it is easy to miss.

ChainedStatefulStreamingTest asserts what the chained pipeline computes.
It cannot distinguish a single query holding two transformWithState
operators from two queries that happen to add up to the same numbers, and
it cannot show whether the watermark moved at all or whether the whole
input simply landed in one micro-batch where everything is trivially on
time.

ChainedStatefulStreamingEvidenceTest asserts the run itself, against
Spark's own StreamingQueryProgress. It pins one record per split per
micro-batch so the watermark has to climb in steps, records every
progress event through a listener, and then asserts that all of them
carry one query id, that one micro-batch reports exactly two
transformWithStateExec state operators, and that the watermark takes at
least three strictly increasing values.

The second test adds a tap between the two operators, so the late record
is observed leaving the dedup operator and absent from the windowed sum.
That is the difference between a record excluded for lateness and a
record that never arrived.

Both tests print the raw per batch progress they recorded, since that
output is the evidence the phase gate report quotes.

Also drops four javadoc references to a Kryo relaxation that no longer
exists anywhere in the suite.